import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.IRichBolt;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Tuple;

import java.util.Map;

/**
 * Created by shuaidong on 2017/7/23.
 */
public class MyRichBolt  implements IRichBolt{

    private OutputCollector outputCollector;

    public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {

        this.outputCollector = outputCollector;
    }

    int num = 0;
    String string = null;

    public void execute(Tuple tuple) {

        //等同于 tuple.getValue(0);
        try {
            string = (String) tuple.getValueByField("log");
            if (string!=null){
                num++;
                System.out.println("lines: "+num+"session_id:"+string.split("\t")[1]);
            }
            outputCollector.ack(tuple);
        } catch (Exception e) {
            //失败重新发送数据
            outputCollector.fail(tuple);

            e.printStackTrace();

        }

    }

    public void cleanup() {

    }

    /**
     * 定义输出字段
     * @param outputFieldsDeclarer
     */
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {

    }

    public Map<String, Object> getComponentConfiguration() {
        return null;
    }
}
